package com.oristartech.netty.common.dispatcher;

import com.oristartech.netty.common.message.NoticeMessage;
import com.oristartech.netty.common.message.RequestMessage;
import com.oristartech.netty.common.message.ResponseMessage;
import com.oristartech.netty.common.util.ValidatorUtils;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;

import java.net.SocketTimeoutException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 ** 通道发送
 *
 * @author: chenfenghai
 * @version: 1.0
 * @date: 2020-03-18 11:00
 * @blame Team
 */
@Slf4j
public class ChannelSender {

    /** 响应结果集，根据请求ID与响应ID一一对应 */
    private static final ConcurrentMap<String, ResponseFuture> RESPONSE_FUTURE_MAP = new ConcurrentHashMap<>();

    /**
     ** 发送通知
     *
     * @param channel 发送通道
     * @param message 通知消息
     */
    public void sendNotice(Channel channel, NoticeMessage message){
        validateChannel(channel);
        ValidatorUtils.validate(message);
        // 发送信息
        channel.writeAndFlush(message.toDataObject());
        log.info("客户端[{}-{}]发送通知：{}",
                message.getTenantId(), message.getCinemaCode(), message);
    }

    /**
     ** 发送请求
     *
     * @param channel 请求通道
     * @param message 请求消息
     * @return ResponseMessage 响应结果
     */
    public ResponseMessage sendRequest(Channel channel, RequestMessage message){
        ResponseMessage responseMessage;
        try {
            responseMessage = executeRequest(channel, message);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        // 返回为null时，表示响应超时
        if(responseMessage == null){
            throw new RuntimeException(new SocketTimeoutException("sendRequest timeout"));
        }
        return responseMessage;
    }

    /**
     ** 发送请求
     *
     * @param channel 请求通道
     * @param message 请求消息
     * @return ResponseMessage 响应结果
     * @throws InterruptedException CountDownLatch.await
     */
    private ResponseMessage executeRequest(Channel channel, RequestMessage message) throws InterruptedException {
        validateChannel(channel);
        ValidatorUtils.validate(message);
        // 使用double checked的方式实现
        // 第一次检查数据是否存在
        ResponseFuture future = RESPONSE_FUTURE_MAP.get(message.getId());
        if(future != null){
            throw new IllegalArgumentException();
        }
        synchronized (this) {
            // 第二次检查数据是否存在
            future = RESPONSE_FUTURE_MAP.get(message.getId());
            if(future != null){
                throw new IllegalArgumentException();
            }
            // 初始化响应监听，并添加到RESPONSE_FUTURE_MAP中
            ResponseFuture responseFuture = new ResponseFuture(this, message);
            future = RESPONSE_FUTURE_MAP.putIfAbsent(message.getId(), responseFuture);
            if(future == null){
                future = responseFuture;
            }
        }
        // 发送信息
        channel.writeAndFlush(message.toDataObject());
        log.info("客户端[{}-{}]发送请求：{}",
                message.getTenantId(), message.getCinemaCode(), message);
        return future.get();
    }

    /**
     ** 接收响应消息，根据响应ID，找到对应的ResponseFuture，并把响应结果保存，最后从RESPONSE_FUTURE_MAP中移除该响应结果
     *
     * @param response 响应消息
     */
    public void received(ResponseMessage response){
        ResponseFuture responseFuture = RESPONSE_FUTURE_MAP.get(response.getId());
        if(responseFuture == null){
            return;
        }
        try {
            responseFuture.received(response);
        }finally {
            RESPONSE_FUTURE_MAP.remove(response.getId());
        }
    }

    /**
     ** 根据请求唯一KEY，移除ResponseFuture
     *
     * @param requestUniqueKey 请求唯一KEY
     */
    public void remove(String requestUniqueKey){
        ResponseFuture responseFuture = RESPONSE_FUTURE_MAP.get(requestUniqueKey);
        if(responseFuture == null){
            return;
        }
        RESPONSE_FUTURE_MAP.remove(requestUniqueKey);
    }

    /**
     ** 验证通道可用
     *
     * @param channel 通道
     */
    private void validateChannel(Channel channel){
        if(channel == null || !channel.isActive()){
            throw new IllegalArgumentException("Channel为空或不可用");
        }
    }
}
